package org.jflame.context.filemanager;

import org.jflame.commons.exception.RemoteAccessException;
import org.jflame.commons.file.FileHelper;
import org.jflame.commons.util.ArrayHelper;
import org.jflame.commons.util.IOHelper;
import org.jflame.commons.util.SimpleRetry;
import org.jflame.commons.util.StringHelper;
import org.jflame.context.filemanager.FtpFileManager.FtpCfg;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.apache.commons.net.ftp.FTPFileFilter;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.commons.net.ftp.FTPSClient;

/**
 * ftp文件管理实现,基于commons-net ftp
 * 
 * @author charles.zhang
 */
public class FtpFileManager extends RemoteFileManager<FtpCfg> {

    private FTPClient ftpClient;
    private volatile boolean isLogin = false;

    public FtpFileManager(FtpCfg cfg) {
        super(cfg);
        StringHelper.requireNonEmpty(cfg.getServer(), "no ftp server configured");
        super.protocolPrefixs = new String[] { "ftp://","ftps://" };
        init();
    }

    private void init() {
        FtpCfg ftpCfg = getConfig();
        if (isProtocolStarter(ftpCfg.getServer())) {
            URI serverUri = URI.create(ftpCfg.getServer());
            ftpCfg.setHost(serverUri.getHost());
            if ("ftps".equals(serverUri.getScheme())) {
                ftpCfg.setEnableFtps(true);
            }
        } else {
            ftpCfg.setHost(ftpCfg.getServer());
            String serverURL = String.format("%s://%s:%d", ftpCfg.isEnableFtps() ? "ftps" : "ftp", ftpCfg.getServer(),
                    ftpCfg.getPort());
            ftpCfg.setServer(serverURL);
        }

        // 默认根目录
        if (StringHelper.isEmpty(ftpCfg.getSavePath())) {
            ftpCfg.setSavePath("/");
        } else {
            if (ftpCfg.getSavePath()
                    .charAt(0) != '/') {
                ftpCfg.setSavePath('/' + ftpCfg.getSavePath());
            }
        }

        if (StringUtils.isBlank(ftpCfg.getUser())) {
            ftpCfg.setUser("anonymous");
            ftpCfg.setPassword("");
        }

        if (ftpCfg.isEnableFtps()) {
            ftpClient = new FTPSClient();
        } else {
            ftpClient = new FTPClient();
        }
    }

    /**
     * 连接并登录到ftp服务器
     * 
     * @throws IOException
     */
    public void login() throws IOException {
        if (isLogin) {
            return;
        }
        FtpCfg cfg = getConfig();

        if (cfg.getConnectionTimeout() > 0) {
            ftpClient.setConnectTimeout(cfg.connectionTimeoutInMillis());
        }
        // 每2分钟在控制连接上发送noop命令避免被路由中断
        ftpClient.setControlKeepAliveTimeout(Duration.ofSeconds(120));
        ftpClient.setControlEncoding(cfg.getCharset() == null ? StandardCharsets.UTF_8.name() : cfg.getCharset());
        if (!ftpClient.isConnected()) {
            // 连接,重试3次
            SimpleRetry.exec(() -> {
                ftpClient.connect(cfg.host, cfg.getPort());
                return ftpClient.getReplyCode();
            }, 3, 2000, r -> !FTPReply.isPositiveCompletion(r), ex -> ex instanceof Exception);
        }
        // 登录
        if (ftpClient.login(cfg.getUser(), cfg.getPassword())) {
            isLogin = true;
            logger.info("ftp登录成功,host={}:{},user={}", cfg.host, cfg.getPort(), cfg.getUser());
        } else {
            logger.error("ftp登录失败,host={}:{},user:{},reply={}", cfg.host, cfg.getPort(), cfg.getUser(),
                    ftpClient.getReplyString());
            ftpClient.disconnect();
            throw new IOException("ftp登录失败");
        }
        if (cfg.isPassiveMode()) {
            ftpClient.enterLocalPassiveMode();
        }

        if (cfg.isBinaryTransfer()) {
            ftpClient.setFileType(FTP.BINARY_FILE_TYPE);
        }

    }

    @Override
    public String save(File file, String saveDir, Map<String,String> fileMeta) throws FileAccessException {
        InputStream fileStream = null;
        String destDir = toAbsDir(saveDir);
        try {
            login();
            changeDir(destDir, true);
            // System.out.println("pwd: " + ftpClient.printWorkingDirectory());
            fileStream = Files.newInputStream(file.toPath());
            if (!ftpClient.storeFile(file.getName(), fileStream)) {
                logger.error("ftp上传文件失败,文件名:{},上传目录:{}, 返回错误:{}", file.getName(), destDir, ftpClient.getReplyString());
                throw new FileAccessException(ftpClient.getReplyString());
            }
        } catch (IOException e) {
            throw new FileAccessException("上传文件异常" + file.getName(), e);
        } finally {
            IOHelper.closeQuietly(fileStream);
        }
        return FileHelper.mergePathToUnix(saveDir, file.getName());
    }

    @Override
    public String save(InputStream fileStream, String filePath, Map<String,String> fileMeta, boolean replaceExisting)
            throws FileAccessException {
        // String destDir = toAbsDir(StringHelper.substringBeforeLast(filePath, "/"));
        String[] paths = splitDirAndFilename(filePath);
        try {
            login();
            changeDir(paths[0], true);
            if (!ftpClient.storeFile(paths[1], fileStream)) {
                logger.error("ftp上传文件失败,文件名:{},上传目录:{}, 返回错误:{}", paths[1], paths[0], ftpClient.getReplyString());
                throw new FileAccessException(ftpClient.getReplyString());
            }
        } catch (IOException e) {
            throw new FileAccessException("上传文件异常" + filePath, e);
        } finally {
            IOHelper.closeQuietly(fileStream);
        }
        return filePath;
    }

    @Override
    public InputStream readStream(String filePath) throws FileAccessException {
        // 改用先将流保存到临时文件再生成输入流, 如果用retrieveFileStream需要在流使用完后close再调用completePendingCommand,否则会影响后续命令
        File tmpFile = read(filePath);
        try {
            return Files.newInputStream(tmpFile.toPath());
        } catch (IOException e) {
            throw new FileAccessException(e);
        }
    }

    @Override
    public byte[] readBytes(String filePath) throws FileAccessException {
        String[] paths = splitDirAndFilename(filePath);
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        try {
            login();
            if (!changeDir(paths[0], false)) {
                throw new FileAccessException("文件不存在" + filePath, FileAccessException.FILE_NOT_FOUND);
            }
            if (ftpClient.retrieveFile(paths[1], outputStream)) {
                return outputStream.toByteArray();
            } else {
                throw new FileAccessException(ftpClient.getReplyString());
            }
        } catch (IOException e) {
            throw new FileAccessException("文件读取异常" + filePath, e);
        }
    }

    @Override
    public File read(String filePath) throws FileAccessException {
        String[] paths = splitDirAndFilename(filePath);
        String key = getFileId(filePath);
        Path tmpPath = mappingTmpFile(key);
        // 使用缓存的临时文件
        Optional<File> localCacheFileOpl = getTmpCacheFile(tmpPath);
        if (localCacheFileOpl.isPresent()) {
            return localCacheFileOpl.get();
        }
        OutputStream outStream;
        try {
            login();
            if (!changeDir(paths[0], false)) {
                throw new FileAccessException("文件不存在" + filePath, FileAccessException.FILE_NOT_FOUND);
            }
            outStream = Files.newOutputStream(tmpPath);
            if (ftpClient.retrieveFile(paths[1], outStream)) {
                File f = tmpPath.toFile();
                return f;
            } else {
                throw new FileAccessException(ftpClient.getReplyString());
            }
        } catch (IOException e) {
            throw new FileAccessException("文件读取异常" + filePath, e);
        }
    }

    /**
     * ftp文件下载
     * 
     * @param filePath 要下载的ftp文件. 如果以/结尾filePath表示目录,此时会下载该目录下所有文件
     * @param savePath 要保存的本地文件路径,如果文件已存在抛出异常
     */
    @Override
    public void download(String filePath, Path savePath) throws FileAccessException {
        if (filePath.endsWith("/")) {
            // 下载文件夹所有文件
            OutputStream localOutStream = null;
            if (!Files.exists(savePath) || !Files.isDirectory(savePath)) {
                try {
                    Files.createDirectories(savePath);
                } catch (IOException e) {
                    throw new FileAccessException("无法创建本地文件夹" + filePath);
                }
            }
            try {
                login();
                if (!changeDir(filePath, false)) {
                    throw new FileAccessException(filePath + "文件不存在", FileAccessException.FILE_NOT_FOUND);
                }
                FTPFile[] ftpFiles = ftpClient.listFiles();
                if (ArrayHelper.isNotEmpty(ftpFiles)) {
                    for (FTPFile fk : ftpFiles) {
                        if (logger.isDebugEnabled()) {
                            logger.debug("文件下载:{}", fk.getName());
                        }
                        try {
                            localOutStream = Files.newOutputStream(savePath.resolve(fk.getName()));
                            if (!ftpClient.retrieveFile(fk.getName(), localOutStream)) {
                                logger.error("文件下载失败:{}", fk.getName());
                            }
                        } finally {
                            IOHelper.closeQuietly(localOutStream);
                        }
                    }
                }
            } catch (IOException e) {
                throw new FileAccessException("ftp文件删除异常", e);
            }
        } else {
            File tmpFile = read(filePath);
            try {
                Files.move(tmpFile.toPath(), savePath, StandardCopyOption.REPLACE_EXISTING);
            } catch (IOException e) {
                throw new FileAccessException("ftp文件下载,保存到本地异常" + filePath, e);
            }
            tmpFile.delete();
        }
    }

    @Override
    public int delete(String... fileKeys) throws FileAccessException {
        FtpCfg cfg = getConfig();
        int i = 0;
        try {
            login();
            /*String workingDir = ftpClient.printWorkingDirectory();
            if (logger.isDebugEnabled()) {
                logger.debug("start del ftp file,working dir:{}", workingDir);
            }*/
            for (String fk : fileKeys) {
                if (ftpClient.deleteFile(FileHelper.mergePathToUnix(cfg.getSavePath(), fk))) {
                    i++;
                } else {
                    logger.error("ftp删除文件失败,{}/{}, 返回:{}", cfg.getSavePath(), fk, ftpClient.getReplyString());
                }
            }
        } catch (IOException e) {
            throw new FileAccessException("ftp文件删除异常", e);
        }
        return i;
    }

    @Override
    public int delete(String parent, String[] fileKeys) throws FileAccessException {
        int i = 0;
        try {
            login();
            if (!changeDir(parent, false)) {
                throw new FileAccessException("ftp目录不存在" + parent);
            }
            for (String fk : fileKeys) {
                if (ftpClient.deleteFile(fk)) {
                    i++;
                } else {
                    logger.error("ftp删除文件失败,{}/{}, 返回:{}", parent, fk, ftpClient.getReplyString());
                }
            }
        } catch (IOException e) {
            throw new FileAccessException("ftp删除文件异常", e);
        }
        return i;
    }

    @Override
    public String copy(String srcFilePath, String destFilePath) throws FileAccessException {
        throw new IllegalStateException("ftp不支持文件复制");
    }

    @Override
    public List<String> listFiles(String dir, int maxFiles, boolean isExcludeDir) throws FileAccessException {
        String[] files = null;
        try {
            login();

            String destDir = toAbsDir(dir);

            /*if (!changeDir(destDir, false)) {
                logger.error("ftp列举目录不存在:{}", destDir);
                return null;
            }*/
            if (isExcludeDir) {
                FTPFile[] ftpFiles = ftpClient.listFiles(destDir, new FTPFileFilter() {

                    @Override
                    public boolean accept(FTPFile file) {
                        return !file.isDirectory();
                    }
                });
                if (ftpFiles == null) {
                    return null;
                }
                if (ftpFiles.length == 0) {
                    return Collections.emptyList();
                }

                Stream<String> fileNameStream = Arrays.stream(ftpFiles)
                        .map(FTPFile::getName);
                if (maxFiles > 0 && maxFiles > ftpFiles.length) {
                    return fileNameStream.limit(maxFiles)
                            .collect(Collectors.toList());
                } else {
                    return fileNameStream.collect(Collectors.toList());
                }

            } else {
                files = ftpClient.listNames(destDir);
                if (files == null) {
                    return null;
                }
                if (files.length == 0) {
                    return Collections.emptyList();
                }
                if (maxFiles > 0 && ArrayHelper.isNotEmpty(files) && maxFiles > files.length) {
                    return Arrays.stream(files)
                            .limit(maxFiles)
                            .collect(Collectors.toList());
                }
                return Arrays.asList(files);
            }
        } catch (IOException e) {
            throw new FileAccessException(e);
        }
    }

    /**
     * 取得目录下ftp文件对象
     * 
     * @param dir
     * @param filter 文件过滤器
     * @return
     * @throws RemoteAccessException
     */
    public FTPFile[] listFtpFiles(String dir, FTPFileFilter filter) throws FileAccessException {
        try {
            login();
            String destDir = toAbsDir(dir);
            if (filter == null) {
                return ftpClient.listFiles(destDir);
            } else {
                return ftpClient.listFiles(destDir, filter);
            }
        } catch (IOException e) {
            throw new FileAccessException(e);
        }
    }

    public FTPClient getFtpClient() {
        return ftpClient;
    }

    /**
     * 切换工作目录,如果目录不存在会自动创建
     * 
     * @param destDir 目标目录
     * @param createIfNotExistDir 是否自动创建不存在的目录
     * @return 切换不成功返回false,通常是目录不存在
     * @throws IOException
     */
    public boolean changeDir(String destDir, boolean createIfNotExistDir) throws IOException {
        if (StringHelper.isEmpty(destDir)) {
            return true; // 表示当前目录不用切
        }
        if (!ftpClient.changeWorkingDirectory(destDir)) {
            if (!createIfNotExistDir) {
                return false;
            }
            if (550 == ftpClient.getReplyCode()) {
                // 切根目录,逐级切换,不存在的目录创建
                ftpClient.changeWorkingDirectory("/");
                makeDirsAndCwd(destDir);
            } else {
                return false;
            }
        }
        return true;
    }

    private String[] splitDirAndFilename(String filePath) {
        String[] dirAndFilename = new String[2];
        int lastIndex = filePath.lastIndexOf("/");
        if (lastIndex < 0) {
            dirAndFilename[0] = toAbsDir("");
            dirAndFilename[1] = filePath;
        } else if (lastIndex == 0) {
            dirAndFilename[0] = toAbsDir("");
            dirAndFilename[1] = filePath.substring(lastIndex + 1);
        } else if (lastIndex > 0) {
            dirAndFilename[0] = toAbsDir(filePath.substring(0, lastIndex));
            dirAndFilename[1] = filePath.substring(lastIndex + 1);
        }

        return dirAndFilename;
    }

    private String toAbsDir(String saveDir) {
        String destDir;
        FtpCfg cfg = getConfig();

        if (StringHelper.isEmpty(saveDir)) {
            destDir = cfg.getSavePath();
        } else {
            destDir = FileHelper.mergePathToUnix(cfg.getSavePath(), saveDir);
        }
        return destDir;
    }

    private void makeDirsAndCwd(String destDir) throws IOException {
        String[] parentDirs = destDir.split("/");
        // 切根目录,逐级切换,不存在的目录创建
        ftpClient.changeWorkingDirectory("/");
        for (String dir : parentDirs) {
            if (!ftpClient.changeWorkingDirectory(dir)) {
                if (550 == ftpClient.getReplyCode()) {
                    if (ftpClient.makeDirectory(dir)) {
                        ftpClient.changeWorkingDirectory(dir);
                    } else {
                        throw new IOException("创建目录失败" + destDir);
                    }
                } else {
                    throw new IOException("创建目录失败" + destDir);
                }
            }
        }
    }

    @Override
    public void close() throws IOException {
        if (ftpClient.isConnected()) {
            try {
                ftpClient.abort();
                ftpClient.logout();
            } catch (IOException e) {
                logger.error("注销ftp异常", e);
            } finally {
                try {
                    ftpClient.disconnect();
                } catch (IOException e) {
                    logger.error("断开ftp连接异常", e);
                }
            }
        }
    }

    public static class FtpCfg extends FileManagerCfg {

        private int port = 21;
        private String user;
        private String password;
        private boolean enableFtps = false;
        private boolean passiveMode = false;// 被动模式连接
        private boolean binaryTransfer = true;// 是否使用二进制传输
        private String charset;
        private int connectionTimeout = -1;

        private transient String host;

        void setHost(String _host) {
            host = _host;
        }

        public int getPort() {
            return port;
        }

        public void setPort(int port) {
            this.port = port;
        }

        public String getUser() {
            return user;
        }

        public void setUser(String user) {
            this.user = user;
        }

        public String getPassword() {
            return password;
        }

        public void setPassword(String password) {
            this.password = password;
        }

        public boolean isPassiveMode() {
            return passiveMode;
        }

        public void setPassiveMode(boolean passiveMode) {
            this.passiveMode = passiveMode;
        }

        public boolean isBinaryTransfer() {
            return binaryTransfer;
        }

        public void setBinaryTransfer(boolean binaryTransfer) {
            this.binaryTransfer = binaryTransfer;
        }

        public String getCharset() {
            return charset;
        }

        public void setCharset(String charset) {
            this.charset = charset;
        }

        public int getConnectionTimeout() {
            return connectionTimeout;
        }

        public int connectionTimeoutInMillis() {
            return (int) TimeUnit.SECONDS.toMillis(connectionTimeout);
        }

        public void setConnectionTimeout(int connectionTimeout) {
            this.connectionTimeout = connectionTimeout;
        }

        public boolean isEnableFtps() {
            return enableFtps;
        }

        public void setEnableFtps(boolean enableFtps) {
            this.enableFtps = enableFtps;
        }

    }

}
